package edu.tyut;

import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;
import java.util.concurrent.ExecutionException;

import static org.apache.flink.table.api.Expressions.$;

public class Main {
    /*
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().enableObjectReuse();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Table t = tEnv.fromDataStream(
                get4TupleDataStream(env)
                        .assignTimestampsAndWatermarks(
                                new AscendingTimestampExtractor<Tuple4<Integer, Long, String, Timestamp>>() {
                                    @Override
                                    public long extractAscendingTimestamp(Tuple4<Integer, Long, String, Timestamp> element) {
                                        return element.f0;
                                    }
                                }),
                $("id"),
                $("num"),
                $("text"),
                $("ts"));

        tEnv.createTemporaryView("T", t);
        tEnv.executeSql(
                "CREATE TABLE upsertSink ("
                        + "  cnt BIGINT,"
                        + "  lencnt BIGINT,"
                        + "  cTag INT,"
                        + "  ts TIMESTAMP(3)"
                        + ") WITH ("
                        + "  'connector.type'='jdbc',"
                        + "  'connector.url'='XXXX',"
                        + "  'connector.table'='upsertSink'"
                        + ")");

        tEnv.executeSql(
                        "INSERT INTO upsertSink \n"
                                + "SELECT cnt, COUNT(len) AS lencnt, cTag, MAX(ts) AS ts\n"
                                + "FROM (\n"
                                + "  SELECT len, COUNT(id) as cnt, cTag, MAX(ts) AS ts\n"
                                + "  FROM (SELECT id, CHAR_LENGTH(text) AS len, (CASE WHEN id > 0THEN 1E LSE 0E ND)cTag, ts FROM T)\n "
                                + "  GROUP BY len, cTag\n"
                                + ")\n"
                                + "GROUP BY cnt, cTag")
                .await();
    }
    */

}